#include <assert.h>
#include <iostream>

#include <stdlib.h>
#include <pthread.h>

#include "Errors.h"
#include "EventProcessorImp.h"
#include "Message.h"


EventProcessorImp::EventProcessorImp(bool _debug, const char *_dbgMsg) :
	msgQueue(_debug, _dbgMsg),  processorsMap(), 
  debug(_debug), dbgMsg(_dbgMsg), forksRemaining(1), myInterface(this)
{
  // register the DIE message, universal.

  RegisterMessageProcessor(DieMessage::type, &EventProcessorImp::processDieMessage, 0);

  // create the mutexes and cond variables
  pthread_mutex_init(&mutex,NULL);
  pthread_mutex_init(&diedMtx,NULL);
  pthread_cond_init(&died,NULL);

  finishedSpins=0;
  exitSpin=false;
}

EventProcessorImp::~EventProcessorImp() {

  // stop all the threads that might be still running
	Seppuku();
//	WaitForProcessorDeath();

	// one last lock just in case this runs too fast
  pthread_mutex_lock(&mutex);
  for(unsigned int i=0; i<vThreads.size(); i++) {
//    pthread_cancel(*(vThreads[i]));

//Alin: this crates memory leaks but I'm fed up trying to avoid double
//free
// delete vThreads[i];
    vThreads[i] = NULL;
  }

  vThreads.clear();
  pthread_mutex_unlock(&mutex);

  // distroy the mutex
  pthread_mutex_destroy(&mutex);
  pthread_mutex_destroy(&diedMtx);
  pthread_cond_destroy(&died);

}

void EventProcessorImp::RegisterMessageProcessor(int Type, msgProcessor Proc, int Priority) {
  processorsMap[Type]=Proc; // put the method in
  msgQueue.AddMessageType(Type, Priority);
}

void EventProcessorImp::Spin() {

  // repeat forever
  while (!exitSpin) {
    // get new message from the queue (blocks until a message is available)

    Message& cMessage = msgQueue.RemoveMessage();
    if (!exitSpin){ // just in case since we might be in the killing phase
      // we do not want to process more messages if we are dieing

      map<int, msgProcessor >::iterator cur=processorsMap.find(cMessage.Type());
      
      // we cannot possibly get a message that we do not suppoprt
      assert(cur!=processorsMap.end());
      
      // get the correct processor
      msgProcessor proc=cur->second;
      
      // call the processor
      (proc)(*this, cMessage);
    }
    
    // this part is separated from the rest for the situation when 
    // the last processed message was a Die message
    if (!exitSpin){ // if we did not die
      // delete the message
      // COMENTED OUT TO DEBUG MEM ALLOCATION      
      delete (&cMessage);
    } else {
      // if we are getting killed, we resend the kill message to make sure
      // everybody gets out of the Spin() method.
      msgQueue.InsertMessage(cMessage);
    } // the last Die message(s) is not pulled form the message queue but that 
    //is fine since we are dying anyway
  }

  // when spin exits we need to determine if this is the last Spin()
  // if it is, we signal all threads waiting for the EventProcessorImp to die
  pthread_mutex_lock(&mutex);
  finishedSpins++;
  if( finishedSpins == vThreads.size() ){
    // we signal all the threads waiting for us to die
    pthread_cond_broadcast(&died);
    
    if (debug){
      printf("EvProc (%s): All Spins() exited. Unblocking waiting threads \n", dbgMsg);
    }
  }
  pthread_mutex_unlock(&mutex);
}

void EventProcessorImp::ProcessMessage(Message& msg) {
    msgQueue.InsertMessage(msg);
}

void EventProcessorImp::processDieMessage(EventProcessorImp &_obj, Message &_msg) {
  //kill all the threads executing for this object
  //stop all the threads that might be still running

  // ensure everybody gets out of the Spin loop as soon as possible
  _obj.exitSpin=true;
  
  if (_obj.debug){
    printf("EvProc Killed(%s): \n", _obj.dbgMsg);
  }
}

void EventProcessorImp::WaitForProcessorDeath(void){
  if (!exitSpin){
    pthread_mutex_lock(&diedMtx);
    pthread_cond_wait(&died, &diedMtx); // no need for a while loop since we want all to wake up
    pthread_mutex_unlock(&diedMtx);
  } // otherwise the EventProcessorImp is dead
}

void EventProcessorImp::Seppuku(){
	DieMessage_Factory(myInterface); // we send ourselved a die message
}


void* EventProcessorImp::ForkAndSpinThread(void* aux){
  // convert the argument to EventProcessorImp
  EventProcessorImp* evProcPtr=(EventProcessorImp*) aux;
  
  // call Spin on the event processor
  evProcPtr->Spin();
  
  pthread_exit(NULL);
  return NULL;
}

bool EventProcessorImp::ForkAndSpin(int node){
  // first, can we even start a thread
  pthread_mutex_lock(&mutex);
  if (forksRemaining == 0){
    pthread_mutex_unlock(&mutex);
    return false;
  }

  // yes we can so decrement the number of threads allowed in the future
  forksRemaining--;

  // create the thread datastructure then call the thread creation.
  pthread_t *threadPtr = new pthread_t;

  // bookkeeping for threads started
  vThreads.push_back(threadPtr);

  pthread_mutex_unlock(&mutex);

  // NULL as the second argument might need to be changed to finetune the starting
  // of the thread
  pthread_attr_t t_attr;
  int ret;

  ret = pthread_attr_init(&t_attr);
  ret = pthread_attr_setdetachstate(&t_attr, PTHREAD_CREATE_DETACHED);
  ret = pthread_create(threadPtr, &t_attr, EventProcessorImp::ForkAndSpinThread, (void *)this);
  FATALIF(ret, "ERROR: return code from pthread_create() is %d\n", ret);

#ifdef USE_NUMA
	// Do we need to pin the thread to a NUMA node?
	if (node!=NUMA_ALL_NODES){
		// wrap the number of nodes arround
		node = node % numaNodeCount();

		// first figure out the cpu's that correspond to this node
		cpu_set_t cpus[MAX_CPU_SETS]; // this allows 640 cpus
		numa_nodes_to_cpus(node, cpus, MAX_CPU_SETS);
		
		// pin the thread down
		pthread_setaffinity_np(threadPtr, MAX_CPU_SETS, cpus);
	}
#endif // USE_NUMA

  return true;
}
